package offline;

import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import tlsb.Utils;
import util.IPartition;
import util.RedisCluster;

class OvlpUnit {
	final String uid;
	Set<String> curFlwee = null;
	boolean isPersisted = false;
	boolean isComputed = false;
	int cuts = 0;
	CGAXiafan cga;
	Set<String> comms = null;
	RedisCluster retRedis = null;
	RedisCluster sourceRedis = null;
	IPartition rPart = null;
	IPartition sPart = null;

	public OvlpUnit(String uid, CGAXiafan cga_) {
		this.uid = uid;
		retRedis = cga_.getRetRedis();
		sourceRedis = cga_.getSourceRedis();
		rPart = retRedis.getPartition();
		sPart = sourceRedis.getPartition();
		cga = cga_;

		String computed = sourceRedis.getJedis(sPart.shard(uid)).hget(uid,
				CGAXiafan.COMPUTED_FIELD);
		if (computed != null && computed.equals("1")) {
			isComputed = true;
		} else {
			String mcf = sourceRedis.getJedis(sPart.shard(uid)).hget(uid,
					CGAXiafan.MAXCUTSFIELD);
			if (mcf != null) {
				cuts = Integer.parseInt(mcf);
			}
		}
	}

	public Set<String> getCurFollowee() {
		if (isComputed)
			return new HashSet<String>();

		if (curFlwee != null) {
			return curFlwee;
		} else {
			curFlwee = sourceRedis.getJedis(sPart.shard(uid)).smembers(
					Utils.joinStr(uid, CGAXiafan.SETOFCURFOLLOWEES));
		}

		if (curFlwee == null || curFlwee.size() == 0) {
			curFlwee = sourceRedis.getJedis(sPart.shard(uid)).smembers(
					Utils.joinStr(uid, CGAXiafan.SETOFFOLLOWEES));
		} else {
			isPersisted = true;
		}
		return curFlwee;
	}

	public boolean isPersisted() {
		return isPersisted;
	}

	/**
	 * 将用户的curfollowee写入到redis中区，默认curfollowee在init函数中是没有创建的
	 */
	public void persist() {
		if (getCurFollowee().size() != 0 && !isPersisted) {
			for (String member : curFlwee) {
				sourceRedis.getPipeline(sPart.shard(uid))
						.sadd(Utils.joinStr(uid, CGAXiafan.SETOFCURFOLLOWEES),
								member);
			}
			sourceRedis.synAllPipelines();
			isPersisted = true;
		}
		storeCuts();
	}

	private void storeCuts() {
		sourceRedis.getJedis(sPart.shard(uid)).hset(uid,
				CGAXiafan.MAXCUTSFIELD, Integer.toString(cuts));
		sourceRedis.getJedis(sPart.shard(CGAXiafan.TODOLIST)).zrem(
				CGAXiafan.TODOLIST, uid);
		if (cuts > 1) {
			sourceRedis.getJedis(sPart.shard(CGAXiafan.TODOLIST)).zadd(
					CGAXiafan.TODOLIST, cuts, uid);
		}
	}

	public Set<String> getComs() {
		while (comms == null) {
			comms = sourceRedis.getJedis(sPart.shard(uid)).smembers(
					Utils.joinStr(uid, CGAXiafan.SETOFCOMMUNITIES));
			if (comms == null)
				cga.getFirstCommunity(uid);
		}
		return comms;
	}

	public Set<String> getProcessedComs() {
		return preprocessCommunities(getComs());
	}

	public void cleanup() {
		persist();

		if (!needCompute()) {
			finish();
		} else if (curFlwee != null && curFlwee.size() == 0) {
			computed();
		}
	}

	public void finish() {
		moveHTL();
		computed();
	}

	public void computed() {
		isComputed = true;
		sourceRedis.getJedis(sPart.shard(uid)).hset(uid,
				CGAXiafan.COMPUTED_FIELD, "1");

		getCurFollowee().clear();
		sourceRedis.getJedis(sPart.shard(uid)).del(
				Utils.joinStr(uid, CGAXiafan.SETOFCURFOLLOWEES));
		// 减少在communities中对已完成用户的不必要的检索
		for (String com : getComs()) {
			sourceRedis.getJedis(sPart.shard(com)).srem(
					Utils.joinStr(uid, CGAXiafan.SETOFCOMMUNITIES), getUid());
		}
		cuts = 0;
		storeCuts();
	}

	public void unComputed() {
		isComputed = false;
		sourceRedis.getJedis(sPart.shard(uid)).hset(uid,
				CGAXiafan.COMPUTED_FIELD, "0");
	}

	public void remElem(Set<String> uids) {
		for (String member : uids) {
			if (isPersisted) {
				sourceRedis.getPipeline(sPart.shard(uid))
						.srem(Utils.joinStr(uid, CGAXiafan.SETOFCURFOLLOWEES),
								member);
			}
			// sourceRedis.getPipeline(sPart.shard(uid)).srem(
			// Utils.joinStr(member, DataLoader.SETOFFOLLOWERS), uid);
		}
		getCurFollowee().removeAll(uids);
		sourceRedis.synAllPipelines();
	}

	public String addToSTL(String sid, Set<String> members) {
		String cid = cga.getFirstCommunity(getUid());
		if (sid == null) {
			sid = cga.addSTL(members, cid);
		}

		// 001:own s1 s2 record stls one user owns
		cga.getIfrMap()
				.get(rPart.shard(cid))
				.writeln(
						"s\t" + Utils.joinStr(getUid(), CGAXiafan.SETOFOWNEDTL)
								+ "\t" + sid);
		retRedis.getJedis(rPart.shard(cid)).sadd(
				Utils.joinStr(getUid(), CGAXiafan.SETOFOWNEDTL), sid);
		decreCuts();
		remElem(members);
		return sid;
	}

	public String getUid() {
		return uid;
	}

	public boolean needCompute() {
		return cuts > 1 && this.getCurFollowee().size() > 1;
	}
	
	public int getCuts() {
		return cuts;
	}

	public void decreCuts() {
		cuts--;
	}

	public boolean isComputed() {
		return isComputed;
	}

	/**
	 * 把uid的curfollowees作为一个share timeline处理
	 */
	public void moveHTL() {
		Set<String> members = getCurFollowee();
		if (members.size() > 0) {
			addToSTL(null, members);
		}
	}

	public static Set<String> preprocessCommunities(Set<String> cidset) {
		List<String> cidList = new LinkedList<String>();
		int cidcounter = 0;
		for (String cid : cidset) {
			cidcounter++;
			if (cidcounter == 1) {
				cidList.add(cid);
				continue;
			}
			if (cidcounter == 2) {
				if (cid.startsWith(cidList.get(0))) {
					cidList.remove(0);
					cidList.add(cid);
					continue;
				}
				if (!cidList.get(0).startsWith(cid)) {
					cidList.add(cid);
				}
			}
			if (cidcounter == 3) {
				for (Iterator<String> ecidIter = cidList.iterator(); ecidIter.hasNext();) {
					String ecid = ecidIter.next();
					if (cid.startsWith(ecid)) {
						ecidIter.remove();
					}
				}
				boolean flag = true;
				for (String ecid : cidList) {
					if (ecid.startsWith(cid)) {
						flag = false;
					}
				}
				if (flag) {
					cidList.add(cid);
				}
			}
		}
		return new HashSet<String>(cidList);
	}
}